package com.snopy.concurrency.demo_threadpool;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
  *@description:自定义线程池
  *@author:tangxiuliang
  *@date:2020/9/25 16:56
  */
@Slf4j(topic = "ThreadPool")
public class ThreadPool {

    private BlockingQueue<Runnable> taskQueue;
    /**线程集合*/
    private final HashSet<Worker> workers = new HashSet<>();
    /**核心线程数*/
    private int  coreSize;
    /**获取任务的超时时间*/
    private long timeout;
    private TimeUnit timeUnit;

     ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
    }
     void execute(Runnable task){
        if (task==null){
            throw new NullPointerException("task can't is  null");
        }
        log.info("workers size:{}",workers.size());
        synchronized (workers){
            if (workers.size() < coreSize){
                log.info("队列任务较少，直接放入核心线程执行：{}",task);
                Worker worker = new Worker(task);
                workers.add(worker);
                worker.start();
            }else{
                log.info("线程池用完，放入阻塞队列：{}",task);
                taskQueue.put(task);
            }
        }
    }

    class Worker extends Thread{
        private Runnable task;
        Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            while (task!=null||(task=taskQueue.take())!=null){
                try {
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            log.info("移除工作线程");
            workers.remove(this);
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {return true;}
            if (!(o instanceof Worker)) {return false;}
            Worker worker = (Worker) o;
            return Objects.equals(task, worker.task);
        }

        @Override
        public int hashCode() {
            return Objects.hash(task);
        }
    }



}


class BlockingQueue<T>{
    private static final Logger log = LoggerFactory.getLogger(BlockingQueue.class);
    BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /**1.任务队列*/
    private Deque<T> deque = new ArrayDeque<>();
    /**2.锁*/
    private ReentrantLock lock = new ReentrantLock();
    /**3生产者条件变量*/
    private Condition fullWaitSet  = lock.newCondition();
    /**消费者条件变量*/
    private Condition emptyWaitSet = lock.newCondition();
    /**容量*/
    private int capacity;
    /**带超时的获取任务*/
    public T poll(long timeout, TimeUnit timeUnit){
        //获取锁
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (deque.isEmpty()){
                //如果等待时间用完，就返回
                if (nanos<=0){
                    return null;
                }
                try {
                     //虚假唤醒问题
                     nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = deque.removeFirst();
            fullWaitSet.signal();
            return t;

        }finally {
            lock.unlock();
        }
    }


    /**
      *@desciption:获取任务方法
      *@pararm:
      *@return:
      *@author:tangxiuliang
      *@date:2020/9/25 16:59
      */

    public T take(){
        //获取锁
        lock.lock();
        try {
            while (deque.isEmpty()){
                try {
                    log.info("没活了，请等待。。。");
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = deque.removeFirst();
            log.info("取出一个任务：{} 任务队列长度{}",t,getSize());
            fullWaitSet.signal();
            return t;

        }finally {
            lock.unlock();
        }

    }
    /**
      *@desciption:添加任务
      *@pararm:
      *@return:
      *@author:tangxiuliang
      *@date:2020/9/25 16:59
      */

    public void put(T task){
        //首先获得锁
        lock.lock();
        try {
            while (getSize()==capacity){
                try {
                    log.info("活太多了，干不动了。。。");
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.info("来活了：{} 起来干活",task);
            deque.addLast(task);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }

    }
    //获取队列长度
    public int getSize(){
        lock.lock();
        try {
            return deque.size();
        }finally {
            lock.unlock();
        }
    }


}
